package com.atguigu.gmall.app.dws.video;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.bean.TableTwoBean;
import com.atguigu.gmall.bean.VideoBean;
import com.atguigu.gmall.bean.VideoBeanTwo;
import com.atguigu.gmall.utils.*;
import io.debezium.data.Json;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.sql.SQLException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class DwsChapterPlay {
    public static void main(String[] args) throws SQLException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        String topicName = "dwd_traffic_page_log";
        String groupID = "channel_flow_app";
        DataStreamSource<String> stringDataStreamSource = env.addSource(KafkaUtil.getFlinkConsumer(topicName, groupID));
        StreamTableEnvironment table = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<VideoBean> jsonObjectSingleOutputStreamOperator = stringDataStreamSource.flatMap(new FlatMapFunction<String, VideoBean>() {
            @Override
            public void flatMap(String value, Collector<VideoBean> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                if(jsonObject.getJSONObject("appVideo")!=null){

                    String string = jsonObject.getJSONObject("appVideo").getString("video_id");
                    DruidDataSource dataSource = DruidPhoenixDSUtil.getDataSource();
                    List<JSONObject> dimInfo = DimUtil.getDimInfo(dataSource.getConnection(), "DIM_VIDEO_INFO", new Tuple2<String, String>("id", string));
                    jsonObject.getJSONObject("appVideo").put("course_id", dimInfo.get(0).getString("chapterId"));
                    JSONObject appVideo = jsonObject.getJSONObject("appVideo");


                    out.collect(new VideoBean(null,
                            null,
                            appVideo.getLong("play_sec"),
                            appVideo.getString("course_id"),
                            0L,
                            jsonObject.getJSONObject("common").getString("mid"),
                            new ArrayList<>(),
                            0L,
                            System.currentTimeMillis()));
                }

            }
        });


        SingleOutputStreamOperator<VideoBean> videoBeanSingleOutputStreamOperator = jsonObjectSingleOutputStreamOperator.assignTimestampsAndWatermarks(WatermarkStrategy.<VideoBean>forBoundedOutOfOrderness(Duration.ofSeconds(1L)).withTimestampAssigner((bean, ts) -> bean.getTs()));
        SingleOutputStreamOperator<VideoBeanTwo> reduce = videoBeanSingleOutputStreamOperator.keyBy(bean -> bean.getCourseId()).window(TumblingEventTimeWindows.of(Time.seconds(3L)))
                .reduce(new ReduceFunction<VideoBean>() {

            @Override
            public VideoBean reduce(VideoBean value1, VideoBean value2) throws Exception {
                if(!value1.getList().contains(value1.getMid())){
                    value1.getList().add(value1.getMid());
                }
                if(!value1.getList().contains(value2.getMid())){
                    value1.getList().add(value2.getMid());
                }

                value1.setPlaySec(value1.getPlaySec() + value2.getPlaySec());
                value1.setCount(value1.getCount() + 1L);
                return value1;
            }
        }, new WindowFunction<VideoBean, VideoBeanTwo, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<VideoBean> input, Collector<VideoBeanTwo> out) throws Exception {
                VideoBeanTwo videoBeanTwo = new VideoBeanTwo();
                VideoBean next = input.iterator().next();

                videoBeanTwo.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                videoBeanTwo.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                videoBeanTwo.setCount(next.getCount());
                videoBeanTwo.setContMind(next.getList().size());
                videoBeanTwo.setPlaySec(next.getPlaySec());
                videoBeanTwo.setCourseId(next.getCourseId());
                videoBeanTwo.setTs(next.getTs());
                if(next.getList().size() != 0){
                    videoBeanTwo.setAvgMindTime(next.getPlaySec()/next.getList().size());
                }else{
                    videoBeanTwo.setAvgMindTime(0L);
                }

                out.collect(videoBeanTwo);
            }
        });

        reduce.print();
        String sql = "insert into dws_chapter_play values(?,?,?,?,?,?,?,?)";
        reduce.addSink(ClickHouseUtil.<VideoBeanTwo>getJdbcSink(sql));

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
